package com.vehicle.mq.worker;

import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.vehicle.common.redis.manager.CacheManager;
import com.vehicle.common.util.thread.pool.ThreadPoolManager;
import com.vehicle.mq.event.EventPublishThread;
import com.vehicle.mq.event.MessageEventHandler;
import com.vehicle.mq.message.BaseMessage;
import com.vehicle.mq.message.MessageEvent;
import com.vehicle.mq.message.MessageEventFacory;
import com.vehicle.mq.queue.EventQueue;

/**
 * Created by yantingbin on 2017/12/5.
 */
public class EventWorker {
    private final Disruptor<MessageEvent> DISRUPTOR;
    private final int RING_BUFFER_SIZE = 1024 * 1024;
    private final EventQueue EVENT_QUEUE;

    public EventWorker(EventQueue eventQueue) {
        this.EVENT_QUEUE=eventQueue;
        DISRUPTOR = new Disruptor<MessageEvent>(new MessageEventFacory(), RING_BUFFER_SIZE, ThreadPoolManager.I.getEventWorkerExecutor(), ProducerType.MULTI, new YieldingWaitStrategy());
    }

    public void init() {
        DISRUPTOR.handleEventsWith(new MessageEventHandler());

        DISRUPTOR.start();

        EventPublishThread eventPublishThread = new EventPublishThread(EVENT_QUEUE);

        eventPublishThread.make(DISRUPTOR.getRingBuffer());
    }
}
